package tag

import java.sql.Connection

import ch.hsr.geohash.GeoHash
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.Row

object BusinessTags extends Tags {
    override def makeTags(args: Any*): List[(String, Int)] = {

      var list = List[(String, Int)]()
      // 参数解析
      val row = args(0).asInstanceOf[Row]
      val conn = args(1).asInstanceOf[Connection]
      val closed = conn

      val lat = row.getAs[String]("lat")
      val long = row.getAs[String]("long")

      println("----------------------------------")
      if (StringUtils.isNotEmpty(lat) && StringUtils.isNotEmpty(long)) {
        val geoHash = GeoHash.withCharacterPrecision(lat.toDouble, long.toDouble, 8).toBase32

        list :+= ("GC"+geoHash, 1)

        val pstmt = conn.prepareStatement("select * from procity_33 where geoHash=?")
        pstmt.setString(1, geoHash)

        val resultSet = pstmt.executeQuery()
        while (resultSet.next()) {
          val bss = resultSet.getString("business")
          if (StringUtils.isNotEmpty(bss)) list :+= ("BS"+bss, 1)
        }
      }

      list

  }
}
